package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{DoubleAccumulator, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

import java.util


object Demo19Code {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("code")
    val sc = new SparkContext(conf)

    //读取分数表
    val scoresRDD: RDD[(String, String, Double)] = sc
      .textFile("data/score.txt")
      .map(_.split(","))
      .map { case Array(id, cid, score) => (id, cid, score.toDouble) }


    //读取学生表
    val studentRDD: RDD[(String, String, Int, String, String)] = sc
      .textFile("data/students.txt")
      .map(_.split(","))
      .map { case Array(id, name, age, sex, clazz) =>
        (id, name, age.toInt, sex, clazz)
      }

    //读取科目表
    val subjectRDD: RDD[(String, String, Double)] = sc
      .textFile("data/subject.txt")
      .map(_.split(","))
      .map { case Array(cid, cname, maxScore) => (cid, cname, maxScore.toDouble) }


    //2.统计总分年级排名前十学生各科的分数
    //计算总分
    val sumScoreRDD: RDD[(String, Double)] = scoresRDD
      .map { case (id, _, score) => (id, score) }
      .reduceByKey((x, y) => x + y)

    //取前十
    val top10: Array[String] = sumScoreRDD
      .sortBy(_._2, ascending = false)
      .take(10)
      .map(_._1)

    //获取各科的分数
    scoresRDD
      .filter { case (id, cid, score) => top10.contains(id) }
      .foreach(println)


    //3.统计总分大于年级平均分的学生
    sumScoreRDD.cache()
    //定义累加器统计学生的人数
    val num: DoubleAccumulator = sc.doubleAccumulator
    val sumScore: Double = sumScoreRDD.map {
      case (id, score) =>
        //统计人数
        num.add(1)
        score
    }.sum()
    val avgScore: Double = sumScore / num.value

    sumScoreRDD
      .filter { case (id, score) => score > avgScore }
      .foreach(println)

    //4.统计每科都及格的学生
    //关联科目表获取总分，计算及格的分数线

    val subjectKvRDD: RDD[(String, Double)] = subjectRDD
      .map { case (cid, cname, maxScore) => (cid, maxScore) }

    val joinRDD: RDD[(String, ((String, Double), Double))] = scoresRDD
      .map { case (id, cid, score) => (cid, (id, score)) }
      .join(subjectKvRDD)

    //整理数据
    val dataRDD: RDD[(String, String, Double, Double)] = joinRDD
      .map {
        case (cid: String, ((id: String, score: Double), maxScore: Double)) =>
          (id, cid, score, maxScore)
      }

    //取出分数及格学生分数
    val jigeRDD: RDD[(String, String, Double, Double)] = dataRDD
      .filter {
        case (id, cid, score, maxScore) => score >= maxScore * 0.6
      }

    //计算及格的科目数
    val allIds: Array[String] = jigeRDD
      .map {
        case (id, cid, score, maxScore) => (id, 1)
      }
      .reduceByKey(_ + _)
      .filter { case (id, num) => num == 6 }
      .collect()
      .map { case (id, num) => id }

    //取出各科的分数
    jigeRDD.filter {
      case (id, cid, score, maxScore) => allIds.contains(id)
    }.foreach(println)
  }
}